package com.flink.streaming.web.webSocket;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.flink.streaming.web.service.AzKaBanSqlService;
import com.flink.streaming.web.thread.AzkabanFlowLogThread;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
public class FlowLogHandler extends TextWebSocketHandler {

    private AzKaBanSqlService azKaBanSqlService;

    private ScheduledExecutorService executorService;

    private ConcurrentHashMap<WebSocketSession, ScheduledFuture> futureMap;

    public FlowLogHandler(AzKaBanSqlService azKaBanSqlService, ScheduledExecutorService executorService, ConcurrentHashMap<WebSocketSession, ScheduledFuture> futureMap) {
        this.futureMap = futureMap;
        this.executorService = executorService;
        this.azKaBanSqlService = azKaBanSqlService;
    }


    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        String data = message.getPayload();
        log.info("入参：{}", data);
        List<String> execIds = conversionData(data);

        AzkabanFlowLogThread logThread = new AzkabanFlowLogThread(1300, execIds, azKaBanSqlService, session, futureMap);
        ScheduledFuture<?> fixedDelay = executorService.scheduleWithFixedDelay(logThread, 0, 850, TimeUnit.MILLISECONDS);
        futureMap.put(session, fixedDelay);
    }


    private List<String> conversionData(String data) {
        JSONObject parseObject = JSONObject.parseObject(data);
        JSONArray execIdsJson = parseObject.getJSONArray("execIds");
        return execIdsJson.toJavaList(String.class);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("长连接关闭");
        ScheduledFuture future = futureMap.get(session);
        if (Objects.nonNull(future)) {
            future.cancel(false);
            futureMap.remove(session);
        }
    }
}
